package flink_p1

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, createTypeInformation}

object FlinkTest_07_Operator_iterator {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(1)

    val dataStream: DataStream[Long] = env.generateSequence(0, 10)


    /**
     * 迭代iterate  第一个参数是一个函数，返回值两个dataStream
     * 第一个dataStream => 继续回到iterate 再经过迭代逻辑
     * 第二个dataStream => 继续回到iterate 执行后续流
     *
     */

    val res: DataStream[Long] = dataStream.iterate(stream => {
      val body: DataStream[Long] = stream.map(x => {
        println("for ==> " + x)
        x > 0 match {
          case true => x - 1
          case false => x
        }
      })
      (body.filter(_ > 0), body.filter(_ <= 0))
    })

    res.print()


    env.execute("test...")

  }
}
